package com.flx.dis.mq.rocket.service.producer;

import com.flx.common.utils.json.JsonUtils;
import com.flx.dis.mq.rocket.entity.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

/**
 * @Author: Fenglixiong
 * @Date: 2021/6/23 11:43
 * @Description:
 *
 * RocketMQ支持三种发送消息的方式
 *
 * 单向发送：
 *      发送端发送完消息后会立即返回，不会等待来自broker的ack来告知本次消息发送是否完全完成发送。
 *      这种方式吞吐量很大，但是存在消息丢失的风险，所以其适用于不重要的消息发送，比如日志收集。
 * 同步发送：
 *      只有在消息完全发送完成之后才返回结果，此方式存在需要同步等待发送结果的时间代价。
 *      具有内部重试机制，即在主动声明本次消息发送失败之前，内部实现将重试一定次数，默认为2次
 * 异步发送：
 *      消息发送后立刻返回，当消息完全完成发送后，会调用回调函数sendCallback来告知发送者本次发送是成功或者失败。
 *      异步模式通常用于响应时间敏感业务场景，即承受不了同步发送消息时等待返回的耗时代价。
 *
 */
@Slf4j
@Component
public class RocketProducer {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 单向发送消息
     * @param message
     */
    public void oneSend(Message message){
        log.info("send message : {}", JsonUtils.toJsonMsg(message));
        rocketMQTemplate.sendOneWay(message.getTopic()+":"+message.getTags(),
                MessageBuilder.withPayload(message.getContent()).build());
    }

    /**
     * 同步发送消息
     * @param message
     */
    public SendResult synSend(Message message){
        log.info("send sync message : {}", JsonUtils.toJsonMsg(message));
        SendResult sendResult = rocketMQTemplate.syncSend(message.getTopic() + ":" + message.getTags(),
                MessageBuilder.withPayload(message.getContent()).build());
        log.info("send sync success : {}",JsonUtils.toJsonMsg(sendResult));
        return sendResult;
    }

    /**
     * 异步发送消息
     * @param message
     */
    public void asyncSend(Message message){
        log.info("send async message : {}", JsonUtils.toJsonMsg(message));
        rocketMQTemplate.asyncSend(message.getTopic() + ":" + message.getTags(), message.getContent(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("send async success : {}",JsonUtils.toJsonMsg(sendResult));
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("send async error : {}",JsonUtils.toJsonMsg(throwable.getMessage()));
            }
        });
    }

    /**
     * 发送事务消息
     *
     * 1、由producer发送prepare（半消息）给MQ的broker。MQ会把消息记录到本地，然后回复prepare消息状态给producer。
     * 2、prepare消息发送以后获取发送状态，如果是成功则执行本地业务（本地事务），根据本地事务执行结果手动返回相应状态（RocketMQLocalTransactionState.COMMIT、RocketMQLocalTransactionState.ROLLBACK等）给MQ。
     * 3、如果是COMMIT则说明本地事务执行成功，prepare为可提交状态，MQ收到COMMIT消息就会发送给consumer，然后consumer执行本地业务。如果是ROLLBACK则会删除prepare消息。
     * 4、如果MQ一直没收到返回状态则会启动定时任务检查本地事务状态
     * 5、消费者、生产者的事务各由开发者自己保证。MQ的事务是由MQ保证，这里会根据使用者配置的参数来决定如何执行。
     */
    public TransactionSendResult sendTransMessage(Message message){
        log.info("send trans message : {}", JsonUtils.toJsonMsg(message));
        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(message.getTopic() + ":" + message.getTags(),
                MessageBuilder.withPayload(message.getContent()).build(), null);
        log.info("send trans success : {}",JsonUtils.toJsonMsg(sendResult));
        return sendResult;

    }

}
